package fun.pengzh.demo.kafka.sample3;

import fun.pengzh.demo.kafka.pojo.Foo2;
import org.apache.kafka.clients.admin.NewTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.converter.BatchMessagingMessageConverter;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.kafka.support.converter.StringJsonMessageConverter;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * 多方法监听（消费者）示例
 */
@SpringBootApplication
public class KafkaDemoApplication {

    private final static Logger logger = LoggerFactory.getLogger(KafkaDemoApplication.class);

    private final TaskExecutor exec = new SimpleAsyncTaskExecutor();

    @Autowired
    private KafkaTemplate template;

    private final static CountDownLatch latch = new CountDownLatch(1);

    public static void main(String[] args) throws Exception {
        final ConfigurableApplicationContext context = SpringApplication.run(KafkaDemoApplication.class, args);
        latch.await();
        Thread.sleep(5000);
        context.close();
    }

    @Bean
    public NewTopic topic2() {
        return TopicBuilder.name("topic2").partitions(1).replicas(1).build();
    }

    @Bean
    public NewTopic topic3() {
        return TopicBuilder.name("topic3").partitions(1).replicas(1).build();
    }

    @Bean
    public RecordMessageConverter converter() {
        return new StringJsonMessageConverter();
    }

    @Bean
    public BatchMessagingMessageConverter batchConverter() {
        return new BatchMessagingMessageConverter(converter());
    }

    @KafkaListener(groupId = "fooGroup2", topics = "topic2")
    public void listener1(List<Foo2> foo2List) throws IOException {
        logger.info("listener1 Received: {}", foo2List);
        foo2List.forEach(f -> {
            template.send("topic3", f.getFoo().toUpperCase());
        });
        logger.info("Message send,Hit Enter to commit tx");
        System.in.read();
    }

    @KafkaListener(groupId = "fooGroup3", topics = "topic3")
    public void listener2(List<String> stringList) {
        logger.info("listener2 Received: {}", stringList);
        latch.countDown();
    }

}

